package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentInfo;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;

/**
 * @author jason
 * @create 2022-04-17 14:18
 */
//数据流： web/App -> nginx -> SpringBoot-> mysql -> FlinkAPP => Kafka(ods) -> FlinkApp -> kafka/Phoenix(dwd/dim)
// -> FlinkApp(redis) -> Kafka(dwm) -> FlinkApp -> Kafka (dwm)

//程序：MockDb -> Mysql -> FlinkCDC -> Kafka(ZK) -> BaseDbApp
// -> Kafka/Phoenix(zk/hdfs/hbase) -> OrderWideApp(Redis) -> Kafka -> PaymentWideApp -> Kafka
public class PaymentWideApp {
    public static void main(String[] args) throws Exception {
        // TODO:1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        //2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序
//        //2.1 开启Checkpoint,每隔5秒钟做一次CK
//        env.enableCheckpointing(5000L);
////        设置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
////        设置最大当前ck数量
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
////        设置最小ck间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
//        //2.2 指定CK的一致性语义
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        //2.3 设置任务关闭的时候保留最后一次CK数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup
//        .RETAIN_ON_CANCELLATION);
//        //2.4 指定从CK自动重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
//        //2.5 设置状态后端
//        env.setStateBackend(new FsStateBackend("hdfs://djcluster/flinkCDC"));
//        //2.6 设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        // TODO: 2.读取kafka主题的数据创建流 并转换为JavaBean对象 提取时间戳生成WaterMark
        String groupId = "payment_wide_group";
        String paymentInfoSourceTopic = "dwd_payment_info";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSinkTopic = "dwm_payment_wide";


        DataStreamSource<String> paymentKafkaDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(orderWideSourceTopic, groupId));

        SingleOutputStreamOperator<OrderWide> orderWideDS =
                paymentKafkaDS.map(jsonStr -> JSON.parseObject(jsonStr, OrderWide.class))
                        .assignTimestampsAndWatermarks(
//                        因为最后还是要给一个时间范围 所以还是用增长的方式分配waterMark
                                WatermarkStrategy.<OrderWide>forMonotonousTimestamps()
                                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderWide>() {
                                            @Override
                                            public long extractTimestamp(OrderWide element, long recordTimestamp) {
                                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                                try {
                                                    return sdf.parse(element.getCreate_time()).getTime();
                                                } catch (ParseException e) {
                                                    e.printStackTrace();
//                                            如果报错 用数据创建时间
                                                    return recordTimestamp;
                                                }
                                            }
                                        }));

        SingleOutputStreamOperator<PaymentInfo> paymentInfoDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(paymentInfoSourceTopic, groupId))
                        .map(jsonStr -> JSON.parseObject(jsonStr, PaymentInfo.class))
                        .assignTimestampsAndWatermarks(
                                WatermarkStrategy.<PaymentInfo>forMonotonousTimestamps()
                                        .withTimestampAssigner(new SerializableTimestampAssigner<PaymentInfo>() {
                                            @Override
                                            public long extractTimestamp(PaymentInfo element, long recordTimestamp) {
                                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                                try {
                                                    return sdf.parse(element.getCreate_time()).getTime();
                                                } catch (ParseException e) {
                                                    e.printStackTrace();
                                                    return recordTimestamp;
                                                }
                                            }
                                        })
                        );

//        // TODO: 3.双流JOIN  订单表 JOIN 支付表  订单表在下 支付表在上  -5表示订单表网络延迟灯迟到5秒 下边界  15表示 支付保留15分钟上边界
//        paymentInfoDS.keyBy(PaymentInfo::getOrder_id)
//                .intervalJoin(orderWideDS.keyBy(OrderWide::getOrder_id))
//                .between(Time.seconds(-5)，Time.minutes(15));
        // TODO: 3.双流JOIN  支付表 JOIN 订单表  支付表在下 订单表在上  -15表示支付表迟到15分钟 下边界  5表示 支付保留5秒上边界
        SingleOutputStreamOperator<PaymentWide> paymentWideDS = paymentInfoDS.keyBy(PaymentInfo::getOrder_id)
                .intervalJoin(orderWideDS.keyBy(OrderWide::getOrder_id))
                .between(Time.minutes(-15), Time.seconds(5))
                .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                    @Override
                    public void processElement(PaymentInfo paymentInfo, OrderWide orderWide,
                                               ProcessJoinFunction<PaymentInfo,
                                                       OrderWide, PaymentWide>.Context ctx,
                                               Collector<PaymentWide> out) throws Exception {
                        out.collect(new PaymentWide(paymentInfo, orderWide));
                    }
                });
//打印测试
        paymentWideDS.print("paymentWideDS================>");

        // TODO: 4. 将数据写入Kafka
        paymentWideDS.map(JSONObject::toJSONString)
                .addSink(MyKafkaUtil.getKafkaProducer(paymentWideSinkTopic));

        // TODO: 5.启动任务
        env.execute("PaymentWideApp");

    }
}
